/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.ulss.data625.handler.datadispatch;

import java.util.HashMap;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.log4j.Logger;
import org.apache.log4j.PropertyConfigurator;

/**
 * @date 2015/2/2
 * @author Carlisle
 */
public class SpeedMonitor {
    
    private static final long INTERVAL = 10;//seconds

    private static AtomicLong sysRecvSpeed = new AtomicLong(0);
    private static AtomicLong sysSendSpeed = new AtomicLong(0);
    private static Map<String, AtomicLong> schemaRecvSpeed = new HashMap<String, AtomicLong>();
    private static Map<String, AtomicLong> schemaSendSpeed = new HashMap<String, AtomicLong>();
    
    private static AtomicLong preSysRecvVol = new AtomicLong(0);
    private static AtomicLong preSysSendVol = new AtomicLong(0);
    private static Map<String, AtomicLong> preSchemaRecvVol = new HashMap<String, AtomicLong>();
    private static Map<String, AtomicLong> preSchemaSendVol = new HashMap<String, AtomicLong>();
    
    private static AtomicLong curSysRecvVol = new AtomicLong(0);
    private static AtomicLong curSysSendVol = new AtomicLong(0);
    private static Map<String, AtomicLong> curSchemaRecvVol = new HashMap<String, AtomicLong>();
    private static Map<String, AtomicLong> curSchemaSendVol = new HashMap<String, AtomicLong>();
    
    boolean stopped = false;
    static Logger logger = null;
    
    static {
        PropertyConfigurator.configure("log4j.properties");
        logger = Logger.getLogger(SpeedMonitor.class.getName());
    }
    
    public SpeedMonitor() {
        
    }
    
    public static void startMonitor() {
        
        logger.info("start speed monitor...");
        Timer updater = new Timer();
        updater.scheduleAtFixedRate(new SpeedUpdater(), 0, INTERVAL * 1000);//****************
        logger.info("speed monitor starts to run...");
    }
    
    public void stopMonitor() {
        stopped = true;
    }
    
    public static void incSchemaRecvVol(String schema, long size) {
        if (!curSchemaRecvVol.containsKey(schema)) {
            
            curSchemaRecvVol.put(schema, new AtomicLong(size));
            preSchemaRecvVol.put(schema, new AtomicLong(0));
            curSysRecvVol.addAndGet(size);

            //System.out.println("【【【【NEWRECV】】】】】" + "curvrecv:" + curSchemaRecvVol + "prerecv" + preSchemaRecvVol + "\n" + "presysrecv" + curSysRecvVol.longValue() + "cursysrecv" + preSysRecvVol.longValue());
        } else {
            curSchemaRecvVol.get(schema).addAndGet(size);
            curSysRecvVol.addAndGet(size);

            //System.out.println("【【【【OLDRECV】】】】】" + "curvrecv:" + curSchemaRecvVol + "prerecv" + preSchemaRecvVol + "\n" + "presysrecv" + curSysRecvVol.longValue() + "cursysrecv" + preSysRecvVol.longValue());
        }
    }
    
    public static void incSchemaSendVol(String schema, long size) {
        if (!curSchemaSendVol.containsKey(schema)) {
            curSchemaSendVol.put(schema, new AtomicLong(size));
            preSchemaSendVol.put(schema, new AtomicLong(0));
            curSysSendVol.addAndGet(size);
            //System.out.println("【【【【OLDSEND】】】】】" + "cursend:" + curSchemaSendVol + "presend" + preSchemaSendVol + "\n" + "presyssend" + curSysSendVol.longValue() + "cursysrecv" + preSysSendVol.longValue());
        } else {
            curSchemaSendVol.get(schema).addAndGet(size);
            curSysSendVol.addAndGet(size);
            //System.out.println("【【【【OLDSEND】】】】】" + "curvsend:" + curSchemaSendVol + "presend" + preSchemaSendVol + "\n" + "presyssend" + curSysSendVol.longValue() + "cursysrecv" + preSysSendVol.longValue());
        }
    }
    
    public static Long getSysRecvSpeed() {
        return sysRecvSpeed.longValue();
    }
    
    public static Long getSysSendSpeed() {
        return sysSendSpeed.longValue();
    }
    
    public static Long getSchemaRecvSpeed(String schema) {
        return schemaRecvSpeed.get(schema).longValue();
    }
    
    public static Long getSchemaSendSpeed(String schema) {
        return schemaSendSpeed.get(schema).longValue();
    }
    
    private static class SpeedUpdater extends TimerTask {
        
        @Override
        public void run() {
//            System.out.println("cur sys recv: " + curSysRecvVol.longValue());
//            System.out.println("pre sys recv: " + preSysRecvVol.longValue());
//            System.out.println("cur sys send: " + curSysSendVol.longValue());
//            System.out.println("pre sys send: " + preSysSendVol.longValue());
            long recv = (curSysRecvVol.longValue() - preSysRecvVol.longValue()) / INTERVAL;
            long send = (curSysSendVol.longValue() - preSysSendVol.longValue()) / INTERVAL;
            sysRecvSpeed.set(recv);
            sysSendSpeed.set(send);
            
            logger.info("System receive speed: " + recv + " byte/s|");
            logger.info("System writting speed: " + send + " byte/s|");
            
            preSysRecvVol.set(curSysRecvVol.longValue());
            preSysSendVol.set(curSysSendVol.longValue());
            
            for (String s : curSchemaRecvVol.keySet()) {
                schemaRecvSpeed.put(s, new AtomicLong((curSchemaRecvVol.get(s).longValue() - preSchemaRecvVol.get(s).longValue()) / INTERVAL));
                preSchemaRecvVol.put(s, new AtomicLong(curSchemaRecvVol.get(s).longValue()));
            }
            for (String s : curSchemaSendVol.keySet()) {
                schemaSendSpeed.put(s, new AtomicLong((curSchemaSendVol.get(s).longValue() - preSchemaSendVol.get(s).longValue()) / INTERVAL));
                preSchemaSendVol.put(s, new AtomicLong(curSchemaSendVol.get(s).longValue()));
            }
            logger.info("recv speed of each schema:" + schemaRecvSpeed + "|send speed of each schema:" + schemaSendSpeed);
            
        }
        
    }
}
